package com.andyyan.elasticsearch.client.impl;

import org.apache.commons.lang.math.NumberUtils;
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by yantingxin on 16-9-8.
 */
public class ESClientPool extends BasePoolableObjectFactory<Client> implements InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(ESClientPool.class);

    private int maxActive;

    private long maxWait;

    private int initPoolSize;

    private String clusterName;

    private String hosts;

    private int defaultPort = 9300;

    private GenericObjectPool<Client> pool;

    public ESClientPool() {}

    public ESClientPool(int maxActive, long maxWait, int initPoolSize, String clusterName, String hosts) {
        this.maxActive = maxActive;
        this.maxWait = maxWait;
        this.initPoolSize = initPoolSize;
        this.clusterName = clusterName;
        this.hosts = hosts;
        this.initPool();
    }

    private void initPool() {
        pool = new GenericObjectPool<Client>(this, maxActive, GenericObjectPool.WHEN_EXHAUSTED_BLOCK, maxWait);
        //改为先进先出
        pool.setLifo(false);
        List<Client> clientList = new ArrayList<Client>(5);
        if (initPoolSize > 0) {
            for (int i = 0; i < initPoolSize; i++) {
                clientList.add(getClient());
            }
            for (Client client : clientList) {
                releaseClient(client);
            }
        }
    }

    public Client getClient() {
        try {
            return pool.borrowObject();
        } catch (Exception e) {
            logger.error("ElasticsearchClientPool getClient error", e);
        }
        return null;
    }

    public void releaseClient(Client client) {
        if (client != null) {
            try {
                pool.returnObject(client);
            } catch (Exception e) {
                logger.error("ElasticsearchClientPool releaseClient error", e);
            }
        }
    }

    @Override
    public void destroyObject(Client obj) throws Exception {
        super.destroyObject(obj);
    }

    @Override
    public Client makeObject() throws Exception {
        Map<String, String> settingMap = new HashMap<String, String>();
        settingMap.put("cluster.name", clusterName);
        Settings settings = Settings.builder().put(settingMap).build();
        TransportClient client = TransportClient.builder().settings(settings).build();
        String[] hostArr = hosts.split(";|,");
        for (String host : hostArr) {
            String[] ipPort = host.split(":");
            String ip = ipPort[0];
            int port = ipPort.length == 1 ? defaultPort : NumberUtils.toInt(ipPort[1], defaultPort);
            client.addTransportAddress(new InetSocketTransportAddress(
                    InetAddress.getByName(ip), port));
        }
        return client;
    }


    public void setMaxActive(int maxActive) {
        this.maxActive = maxActive;
    }

    public void setMaxWait(long maxWait) {
        this.maxWait = maxWait;
    }

    public void setInitPoolSize(int initPoolSize) {
        this.initPoolSize = initPoolSize;
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    public void setHosts(String hosts) {
        this.hosts = hosts;
    }

    public void setDefaultPort(int defaultPort) {
        this.defaultPort = defaultPort;
    }

    public void destroy() throws Exception {
        if (pool != null) {
            pool.close();
        }
    }

    public void afterPropertiesSet() throws Exception {
        Assert.notNull(clusterName, "clusterName is required");
        Assert.notNull(hosts, "hosts is required");
    }
}
